package com.xb.loan.util;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;
import redis.clients.jedis.exceptions.JedisException;
import java.io.IOException;
import java.util.*;

/**
 * redis 封装类
 * Created by liuxl on 14/12/11.
 */
public class RedisUtils {
    private static Logger logger = LoggerFactory.getLogger(RedisUtils.class);
    public static ObjectMapper jsonMapper = JacksonJsonMapperUtil.getMapper();

    private JedisPool jedisPool;

    static {
//        jsonMapper = new ObjectMapper();
//        jsonMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
//        jsonMapper.configure(SerializationFeature.WRITE_SINGLE_ELEM_ARRAYS_UNWRAPPED, false);
//        jsonMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        jsonMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

//        objectMapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
    }

    public Long ttl(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.ttl(key);
            }
        }.setNameKey("ttl", key).getResult();
    }

    public boolean setbit(final String key, final long offset, final boolean val) {
        return new Executor<Boolean>(jedisPool) {

            @Override
            Boolean execute() {
                return jedis.setbit(key, offset, val);
            }
        }.setNameKey("setbit", key).getResult();
    }

    public boolean getbit(final String key, final long offset) {
        return new Executor<Boolean>(jedisPool) {

            @Override
            Boolean execute() {
                return jedis.getbit(key, offset);
            }
        }.setNameKey("getbit", key).getResult();
    }

    public Long bitcount(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.bitcount(key);
            }
        }.setNameKey("bitcount", key).getResult();
    }

    public Long setrangeByteArr(final String key, final long offset, final byte[] data) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.setrange(key.getBytes(), offset, data);
            }
        }.setNameKey("setrangeByteArr", key).getResult();
    }

    public byte[] getrangeByteArr(final String key, final long offset, final int size) {
        return new Executor<byte[]>(jedisPool) {

            @Override
            byte[] execute() {
                return jedis.getrange(key.getBytes(), offset, offset + size);
            }
        }.setNameKey("getrangeByteArr", key).getResult();
    }

    public Long delete(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.del(key);
            }
        }.setNameKey("delete", key).getResult();
    }

    public Long mdel(final String... keys) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.del(keys);
            }
        }.setNameKey("mdel", String.valueOf(keys.length)).getResult();
    }

    public Long expire(final String key, final int expire) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                return jedis.expire(key, expire);
            }
        }.setNameKey("expire", key).getResult();
    }

    public Long expireAt(final String key, final long unixTime) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                return jedis.expireAt(key, unixTime);
            }
        }.setNameKey("expireAt", key).getResult();
    }

    public String setString(final String key, final String value) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                return jedis.set(key, value);
            }
        }.setNameKey("setString", key).getResult();
    }

    public String setByteArr(final String key, final byte[] value) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {

                return jedis.set(key.getBytes(), value);
            }
        }.setNameKey("setByteArr", key).getResult();
    }

    public String setByteArrWithSeconds(final String key, final byte[] value, final long expire) {
        return new Executor<String>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            String execute() {
                pipeline = jedis.pipelined();
                Response result = pipeline.set(key.getBytes(), value);
                pipeline.expire(key, (int) expire);
                pipeline.sync();
                return (String) result.get();
            }
        }.setNameKey("setByteArrWithSeconds", key).getResult();
    }


    public Long publish(final String channel, final byte[] message) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {

                return jedis.publish(channel.getBytes(), message);
            }
        }.setNameKey("publish", channel).getResult();
    }

    public String mset(final List<Pair<String, byte[]>> kvs) {
        if (kvs == null || kvs.size() < 1) {
            return null;
        }

        List<String> keys =
                Lists.transform(kvs, new Function<Pair<String, byte[]>, String>() {
                    @Override
                    public String apply(Pair<String, byte[]> pair) {
                        return pair.getLeft();
                    }
                });

        return new Executor<String>(jedisPool) {
            @Override
            String execute() {
                byte[][] byss = new byte[kvs.size() * 2][];

                int idx = 0;
                for (Pair<String, byte[]> kv : kvs) {
                    byss[idx++] = kv.getLeft().getBytes();
                    byss[idx++] = kv.getRight();
                }
                return jedis.mset(byss);
            }
        }.setNameKey("mset", Joiner.on(" ").join(keys)).getResult();
    }

    public String setObject(final String key, final Object obj) throws JsonProcessingException {
        final String jsonValue = jsonMapper.writeValueAsString(obj);
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                return jedis.set(key, jsonValue);
            }
        }.setNameKey("setObject", key).getResult();
    }


    public String setObjectWithSecond(final String key, final Object obj, final long second) throws JsonProcessingException {
        final String jsonValue = jsonMapper.writeValueAsString(obj);
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                Pipeline pipeline = jedis.pipelined();
                Response result = pipeline.set(key, jsonValue);
                pipeline.expire(key, (int) second);
                pipeline.sync();
                return result.get().toString();
            }
        }.setNameKey("setObjectWithSecond", key).getResult();
    }


    public Long setNxWithinSeconds(final String key, final String value, final long expire) {
        return new Executor<Long>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            Long execute() {
                pipeline = jedis.pipelined();
                Response result = pipeline.setnx(key, value);
                pipeline.expire(key, (int) expire);
                pipeline.sync();
                return (Long) result.get();
            }
        }.setNameKey("setNxWithinSeconds", key).getResult();
    }

    public Long hsetAndGetLen(final String key, final String filed, final String value, final int expire) {
        return new Executor<Long>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            Long execute() {
                pipeline = jedis.pipelined();
                pipeline.hset(key, filed, value);
                pipeline.expire(key, expire);
                Response<Long> resp2 = pipeline.hlen(key);
                pipeline.sync();

                return (Long) resp2.get();
            }
        }.setNameKey("hsetAndGetLen", key).getResult();
    }

    public Long hsetnx(final String key, final String filed, final String value) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                return jedis.hsetnx(key, filed, value);
            }
        }.setNameKey("hsetnx", key).getResult();
    }

    public Long hset(final String key, final String filed, final String value, final int expire) {
        return new Executor<Long>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            Long execute() {
                pipeline = jedis.pipelined();
                Response<Long> resp2 = pipeline.hset(key, filed, value);
                pipeline.expire(key, expire);
                pipeline.sync();

                return (Long) resp2.get();
            }
        }.setNameKey("hset", key).getResult();
    }

    public Long hdelAndGetLen(final String key, final String filed) {
        return new Executor<Long>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            Long execute() {
                pipeline = jedis.pipelined();
                Response<Long> result = pipeline.hdel(key, filed);
                Response<Long> resp2 = pipeline.hlen(key);
                pipeline.sync();
                return (Long) resp2.get();
            }
        }.setNameKey("hdelAndGetLen", key).getResult();
    }


    public String setStringWithinSeconds(final String key, final String value, final long expire) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                return jedis.setex(key, (int) expire, value);
            }
        }.setNameKey("setStringWithinSeconds", key).getResult();
    }

    public String setByteWithinSeconds(final String key, final byte[] value, final int expire) {
        return new Executor<String>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            String execute() {

                byte[] keyByte = key.getBytes();

                pipeline = jedis.pipelined();
                Response<String> result = pipeline.set(keyByte, value);
                pipeline.expire(keyByte, expire);
                pipeline.sync();

                return (String) result.get();
            }
        }.setNameKey("setByteWithinSeconds", key).getResult();
    }

    public String setObjectWithinSeconds(final String key, final Object obj, final int timeout) throws JsonProcessingException {
        final String value = jsonMapper.writeValueAsString(obj);

        return new Executor<String>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            String execute() {
                pipeline = jedis.pipelined();
                Response result = pipeline.set(key, value);
                pipeline.expire(key, timeout);
                pipeline.sync();
                return (String) result.get();
            }
        }.setNameKey("setObjectWithinSeconds", key).getResult();
    }


//    /**
//     * 对象以二进制形式存储
//     *
//     * @param key
//     * @param value
//     * @param expire
//     * @return
//     */
//    public String setSerialObjectWithinSeconds(final String key, final Object value, final int expire) {
//        byte[] v = SerializationUtil.serialize(value);
//        return setByteArrWithinSeconds(key, v, expire);
//    }
//
//    /**
//     * 对象以二进制形式存储，无有效期
//     *
//     * @param key
//     * @param value
//     * @return
//     */
//    public String setSerialObject(final String key, final Object value) {
//        byte[] v = SerializationUtil.serialize(value);
//        return setByteArr(key, v);
//    }
//
//
//    /**
//     * 根据key取二进制对象
//     *
//     * @param cacheKey
//     * @param returnClass
//     * @param <T>
//     * @return
//     */
//    public <T> T getSerialObject(final String cacheKey, final Class<T> returnClass) {
//        byte[] b = getByteArr(cacheKey);
//        return SerializationUtil.deserialize(b, returnClass);
//    }


    public <T> String hPutAllObject(final String cacheKey, final Map<String, T> mapData) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {

                Map<byte[], byte[]> objectMap = Maps.newHashMap();
                Set<String> keySet = mapData.keySet();
                for (String key : keySet) {
                    T obj = mapData.get(key);
                    if (obj != null) {
                        objectMap.put(key.getBytes(), SerializeUtil.serialize(obj));
                    }
                }

                return jedis.hmset(cacheKey.getBytes(), objectMap);
            }
        }.setNameKey("hPutAllObject", cacheKey).getResult();
    }

    public Map<byte[], byte[]> hGetAllToByte(final String key) {
        return new Executor<Map<byte[], byte[]>>(jedisPool) {

            @Override
            Map<byte[], byte[]> execute() {
                return jedis.hgetAll(key.getBytes());
            }
        }.setNameKey("hGetAll", key).getResult();
    }

    public <T> String hPutAll(final String cacheKey, final Map<String, String> mapData) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {

                Map<String, String> objectMap = Maps.newHashMap();
                Set<String> keySet = mapData.keySet();
                for (String key : keySet) {
                    String obj = mapData.get(key);
                    if (obj != null) {
                        objectMap.put(key, obj);
                    }
                }

                return jedis.hmset(cacheKey, objectMap);
            }
        }.setNameKey("hPutAll", cacheKey).getResult();
    }

    public Set<String> hkeys(final String cacheKey) {
        return new Executor<Set<String>>(jedisPool) {
            @Override
            Set<String> execute() {
                return jedis.hkeys(cacheKey);
            }
        }.setNameKey("hkeys", cacheKey).getResult();
    }

    public Long hPut(final String cacheKey, final String key, final Object obj) throws Exception {
        final String jsonValue = jsonMapper.writeValueAsString(obj);
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hset(cacheKey, key, jsonValue);
            }
        }.setNameKey("hPut", cacheKey).getResult();
    }

    public Long hlen(final String cacheKey) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hlen(cacheKey);
            }
        }.setNameKey("hlen", cacheKey).getResult();
    }

    public Long hdel(final String cacheKey, final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hdel(cacheKey, key);
            }
        }.setNameKey("hdel", cacheKey).getResult();
    }

    public Long hdel(final String cacheKey, final String... fields) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hdel(cacheKey, fields);
            }
        }.setNameKey("hdel", cacheKey).getResult();
    }


    public Long hPutString(final String cacheKey, final String key, final String str) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hset(cacheKey, key, str);
            }
        }.setNameKey("hPutString", cacheKey).getResult();
    }

    public Long incrBy(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.incrBy(key, 1);
            }
        }.setNameKey("incrBy", key).getResult();
    }

    public Long incrBy(final String key, final int num) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.incrBy(key, num);
            }
        }.setNameKey("incrBy", key).getResult();
    }

    public Long incr(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.incr(key);
            }
        }.setNameKey("incr", key).getResult();
    }

    public Long incrBy(final String key, final Integer num) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.incrBy(key, num);
            }
        }.setNameKey("incr", key).getResult();
    }

    public Long decr(final String key) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                return jedis.decr(key);
            }
        }.setNameKey("decr", key).getResult();
    }


//    String result = operations.execute(new SessionCallback<String>() {
//        @SuppressWarnings({"unchecked", "rawtypes"})
//        public String execute(RedisOperations operations) throws DataAccessException {
//            List<Object> execResult = null;
//            do {
//                operations.watch(key);
//                Boolean result = operations.hasKey(key);
//                operations.multi();
//                BoundValueOperations<String, String> valOperations = operations.boundValueOps(key);
//                if (!result) {
//                    valOperations.increment(seq + 1);
//                } else {
//                    valOperations.increment(1l);
//                }
//            } while ((execResult = operations.exec()) == null);
//            return String.valueOf(execResult.get(0));
//        }
//    });
//    return Long.parseLong(result);
//}

    public Long transactionRedis(final String key, final Long value) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {

                Long result = jedis.setnx(key, String.valueOf(value));
                return jedis.incrBy(key, 1);
            }
        }.setNameKey("transactionRedis", key).getResult();
    }

    public Long transactionRedis(final String key, final Long value, final int num) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {

                Long result = jedis.setnx(key, String.valueOf(value));
                return jedis.incrBy(key, num);
            }
        }.setNameKey("transactionRedis", key).getResult();
    }

    public Long hincrBy(final String cacheKey, final String key) {
        return hincrBy(cacheKey, key, 1);
    }

    public Long hincrBy(final String cacheKey, final String key, final int num) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.hincrBy(cacheKey, key, num);
            }
        }.setNameKey("hincrBy", key).getResult();
    }


    public Long setNx(final String key, final String value) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.setnx(key, value);
            }
        }.setNameKey("setNx", key).getResult();
    }


    public Boolean exists(final String key) {
        return new Executor<Boolean>(jedisPool) {

            @Override
            Boolean execute() {
                return jedis.exists(key);
            }
        }.setNameKey("exists", key).getResult();
    }


    public Map<String, Boolean> multiExists(final List<String> keys) {
        return new Executor<Map<String, Boolean>>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            Map<String, Boolean> execute() {
                pipeline = jedis.pipelined();

                List<Response<Boolean>> resps = Lists.newArrayListWithCapacity(keys.size());
                for (String key : keys) {
                    resps.add(pipeline.exists(key));
                }

                pipeline.sync();

                Map<String, Boolean> result = Maps.newHashMap();
                for (int i = 0; i < keys.size(); i++) {
                    String key = keys.get(i);
                    result.put(key, resps.get(i).get());
                }
                return result;
            }
        }.setNameKey("multiExists", "nkeys:" + keys.size()).getResult();
    }

    public String get(final String key) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                return jedis.get(key);
            }
        }.setNameKey("get", key).getResult();
    }

    public byte[] getSetByteArr(final String key, final byte[] value) {
        return new Executor<byte[]>(jedisPool) {

            @Override
            byte[] execute() {
                return jedis.getSet(key.getBytes(), value);
            }
        }.setNameKey("getSetByteArr", key).getResult();
    }

    public List<byte[]> mget(final List<String> keys) {
        if (keys == null || keys.size() < 1) {
            return null;
        }

        List<byte[]> kbList = Lists.transform(keys, new Function<String, byte[]>() {
            @Override
            public byte[] apply(String key) {
                return key.getBytes();
            }
        });

        final byte[][] kbs = kbList.toArray(new byte[keys.size()][]);

        return new Executor<List<byte[]>>(jedisPool) {
            @Override
            List<byte[]> execute() {
                return jedis.mget(kbs);
            }
        }.setNameKey("mget", Joiner.on(" ").join(keys)).getResult();
    }

    public byte[] getByteArr(final String key) {
        return new Executor<byte[]>(jedisPool) {

            @Override
            byte[] execute() {

                byte[] keyArr = key.getBytes();

                return jedis.get(keyArr);
            }
        }.setNameKey("getByteArr", key).getResult();
    }

    public String hGet(final String key, final String field) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                return jedis.hget(key, field);
            }
        }.setNameKey("hGet", key).getResult();
    }

    public ScanResult<Map.Entry<String, String>> hscan(final String key, final String cursor, final int count) {
        return new Executor<ScanResult<Map.Entry<String, String>>>(jedisPool) {

            @Override
            ScanResult<Map.Entry<String, String>> execute() {
                return jedis.hscan(key, cursor, new ScanParams().count(count));
            }
        }.setNameKey("hscan", key).getResult();
    }


    public String hGet(final String key, final String field, final int expire) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<String> result = pipeline.hget(key, field);
                pipeline.expire(key, expire);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("hGet", key).getResult();
    }


    public String hashMultipleSet(final String key, final Map<String, String> hash, final int expire) {
        return new Executor<String>(jedisPool) {

            @Override
            String execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<String> result = pipeline.hmset(key, hash);
                pipeline.expire(key, expire);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("hashMultipleSet", key).getResult();
    }

    public List<String> hashMultipleGet(final String key, final String... fields) {
        return new Executor<List<String>>(jedisPool) {

            @Override
            List<String> execute() {
                return jedis.hmget(key, fields);
            }
        }.setNameKey("hashMultipleGet", key).getResult();
    }

    public List<String> hashMultipleGet(final String key, final int expire, final String... fields) {
        return new Executor<List<String>>(jedisPool) {

            @Override
            List<String> execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<List<String>> result = pipeline.hmget(key, fields);
                pipeline.expire(key, expire);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("hashMultipleGet", key).getResult();
    }

    public Map<String, String> hGetAll(final String key) {
        return new Executor<Map<String, String>>(jedisPool) {

            @Override
            Map<String, String> execute() {
                return jedis.hgetAll(key);
            }
        }.setNameKey("hGetAll", key).getResult();
    }


    public <T> Map<String, T> hGetAll(final String key, final TypeReference<T> type) throws IOException {
        Map<String, T> object = null;
        Map<String, String> cache = hGetAll(key);
        if (cache != null) {
            object = new HashedMap();


            for (Map.Entry<String, String> entry : cache.entrySet()) {
                T value = (T) jsonMapper.readValue(entry.getValue(), type);
                object.put(entry.getKey(), value);
            }
        }
        return object;
    }
//
//    public <T> List<Map<String, T>> hGetAll(final List<String> key,final TypeReference<T> type) throws IOException  {
//        final List<Map<String, T>> object = null;
//        batchExecuteWithType(new Callback() {
//            @Override
//            public void callback(Pipeline pipeline) {
//                for(String str:key){
//                    Map<String,String> cache = pipeline.hgetAll(str).get();
//                    T value = (T) jsonMapper.readValue(cache.getValue(), type);
//
//                    object.putAll();
//                }
//            }
//        });
//        return object;
//    }


    public Long zadd(final String key, final double score, final byte[] member, final int expire) {
        return addWithSortedSet(key, score, member, expire);
    }

    public Long zadd(final String key, final double score, final byte[] member) {
        return addWithSortedSet(key, score, member);
    }

    public Long zadd(final String key, final double score, final Object object) throws Exception {
        final String member = jsonMapper.writeValueAsString(object);
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.zadd(key, score, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("zadd", key).getResult();
    }


//
//    public Long zrem(final String key, final Object object) throws Exception{
//        final String member = jsonMapper.writeValueAsString(object) ;
//        return new Executor<Long>(jedisPool) {
//            @Override
//            Long execute() {
//                Pipeline pipeline = jedis.pipelined();
//                Response<Long> result = pipeline.zrem(key,member);
//                pipeline.sync();
//                return result.get();
//            }
//        }.setNameKey("zrem", key).getResult();
//    }


    public Long zrem(final String key, final String member) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.zrem(key, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("zrem", key).getResult();
    }

    public Long zcard(final String key) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zcard(key);
            }
        }.setNameKey("zcard", key).getResult();
    }

    public Long zcount(final String key, final double min, final double max) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zcount(key, min, max);
            }
        }.setNameKey("zcard", key).getResult();
    }


    public Long addWithSortedSet(final String key, final double score, final byte[] member, final int expire) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {

                byte[] keyArr = key.getBytes();

                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.zadd(keyArr, score, member);
                pipeline.expire(keyArr, expire);
                pipeline.sync();
                return result.get();

            }
        }.setNameKey("addWithSortedSet", key).getResult();
    }

    public Long addWithSortedSet(final String key, final double score, final byte[] member) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                byte[] keyArr = key.getBytes();

                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.zadd(keyArr, score, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("addWithSortedSet", key).getResult();
    }

    public Long zadd(final String key, final double score, final String member) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.zadd(key, score, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("zadd", key).getResult();
    }

//    public Long zadd(final String key, final Map<String, Double> map) {
//        return new Executor<Long>(jedisPool) {
//
//            @Override
//            Long execute() {
//                byte[] keyArr = key.getBytes();
//                Pipeline pipeline = jedis.getShard(key).pipelined();
////                Response<Long> result = pipeline.zadd(keyArr, score, member);
//                Response<Long> result = pipeline.zadd(key, map);
//                pipeline.sync();
//                return result.get();
//            }
//        }.setNameKey("zadd", key).getResult();
//    }

    public Long zadd(final String key, final Map<byte[], Double> map) {
        return zadd(key, map, 0);
    }

    public Long zadd(final String key, final Map<byte[], Double> map, final int expire) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                Long count = jedis.zadd(key.getBytes(), map);
                if (expire > 0) {
                    jedis.expire(key, expire);
                }
                return count;
            }
        }.setNameKey("zadd", key).getResult();
    }

    public Set<byte[]> revrangeByScoreWithSortedSet(final String key, final double max, final double min) {
        return new Executor<Set<byte[]>>(jedisPool) {

            @Override
            Set<byte[]> execute() {
                return jedis.zrevrangeByScore(key.getBytes(), max, min);
            }
        }.setNameKey("revrangeByScoreWithSortedSet", key).getResult();
    }

    /**
     * 批量查询zset
     *
     * @param keys key数组
     * @param max  最大score
     * @param mins 最小score的数组
     * @return
     */
    public List<Object> revrangeBatchByScoreWithSortedSet(final String[] keys, final double max, final Double[] mins) {
        String geStr = Arrays.toString(keys);   //传给globalEye，显示为key
        if (geStr.length() > 30) {
            geStr = geStr.substring(0, 22) + " ...(" + keys.length + ")";
        }

        return new Executor<List<Object>>(jedisPool) {

            @Override
            List<Object> execute() {
                Pipeline pipeline = jedis.pipelined();
                for (int i = 0; i < keys.length; i++) {
                    pipeline.zrevrangeByScore(keys[i].getBytes(), max, mins[i]);
                }
                return pipeline.syncAndReturnAll();
            }
//        }.setName("revrangeBatchByScoreWithSortedSet").getResult();
        }.setNameKey("revrangeBatchByScoreWithSortedSet", geStr).getResult();
    }

    public Set<Tuple> zrangeWithScores(final String key, final long min, final long max) {
        return new Executor<Set<Tuple>>(jedisPool) {
            @Override
            Set<Tuple> execute() {
                return jedis.zrangeWithScores(key, min, max);
            }
        }.setNameKey("zrange", key).getResult();
    }

    public <T> Set<T> zrange(final String key, final long min, final long max, final TypeReference<T> type) {
        return new Executor<Set<T>>(jedisPool) {
            @Override
            Set<T> execute() {
                Set<byte[]> bs = jedis.zrange(key.getBytes(), min, max);
                Set<T> result = null;
                if (bs != null) {
                    result = new HashSet<>();
                    for (byte[] bytes : bs) {
                        T t = null;
                        try {
                            t = (T) jsonMapper.readValue(bytes, type);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        result.add(t);
                    }
                }
                return result;
            }
        }.setNameKey("zrange", key).getResult();
    }

    public Set<byte[]> zrangeByScore(final String key, final double min, final double max) {
        return new Executor<Set<byte[]>>(jedisPool) {
            @Override
            Set<byte[]> execute() {
                return jedis.zrangeByScore(key.getBytes(), min, max);
            }
        }.setNameKey("zrangeByScore", key).getResult();
    }

    public Set<Tuple> zrangeByScoreWithScores(final String key, final double min, final double max) {
        return new Executor<Set<Tuple>>(jedisPool) {
            @Override
            Set<Tuple> execute() {
                return jedis.zrangeByScoreWithScores(key.getBytes(), min, max);
            }
        }.setNameKey("zrangeByScoreWithScores", key).getResult();
    }

    public Set<Tuple> zrangeByScoreWithScores(final byte[] key, final byte[] min, final byte[] max) {
        return new Executor<Set<Tuple>>(jedisPool) {
            @Override
            Set<Tuple> execute() {
                return jedis.zrangeByScoreWithScores(key, min, max);
            }
        }.setNameKey("zrangeByScoreWithScores", "byteKey").getResult();
    }


    public Set<Tuple> zrangeByScoreWithScores(final String key, final String min, final String max) {
        return new Executor<Set<Tuple>>(jedisPool) {
            @Override
            Set<Tuple> execute() {
                return jedis.zrangeByScoreWithScores(key, min, max);
            }
        }.setNameKey("zrangeByScoreWithScores", key).getResult();
    }


    public Double zscore(final String key, final String member) {
        return new Executor<Double>(jedisPool) {
            @Override
            Double execute() {
                return jedis.zscore(key, member);
            }
        }.setNameKey("zscore", key).getResult();
    }

    public String setByteArrWithinSeconds(final String key, final byte[] value, final int expire) {
        return new Executor<String>(jedisPool) {
            Pipeline pipeline = null;

            @Override
            String execute() {

                byte[] keyByte = key.getBytes();

                pipeline = jedis.pipelined();
                Response<String> result = pipeline.set(keyByte, value);
                pipeline.expire(keyByte, expire);
                pipeline.sync();

                return result.get();
            }
        }.setNameKey("setByteArrWithinSeconds", key).getResult();
    }


    public Long addWithSortedSet(final String key, final Map<String, Double> scoreMembers) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zadd(key, scoreMembers);
            }
        }.setNameKey("addWithSortedSet", key).getResult();
    }

    /**
     * 移除有序集 key 中，指定排名(rank)区间内的所有成员。
     * 区间分别以下标参数 start 和 stop 指出，包含 start 和 stop 在内。
     * 下标参数 start 和 stop 都以 0 为底，也就是说，以 0 表示有序集第一个成员，以 1 表示有序集第二个成员，以此类推。
     * 你也可以使用负数下标，以 -1 表示最后一个成员， -2 表示倒数第二个成员，以此类推。
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public Long zremrangeByRank(final String key, final long start, final long end) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zremrangeByRank(key, start, end);
            }
        }.setNameKey("zremrangeByRank", key).getResult();
    }

    public Long zremByScore(final String key, final Double start, final Double end) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zremrangeByScore(key, start, end);
            }
        }.setNameKey("ZREMRANGEBYSCORE", key).getResult();
    }

    public Long zremByScore(final String key, final long start, final long end) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zremrangeByScore(key, start, end);
            }
        }.setNameKey("ZREMRANGEBYSCORE", key).getResult();
    }

    public Long zrem(final String key, final String... members) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zrem(key, members);
            }
        }.setNameKey("zrem", key).getResult();
    }

    public Long zrem(final String key, final byte[]... members) {
        return new Executor<Long>(jedisPool) {

            @Override
            Long execute() {
                return jedis.zrem(key.getBytes(), members);
            }
        }.setNameKey("zrem", key).getResult();
    }

    public Long sadd(final String key, final String member) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.sadd(key, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("sadd", key).getResult();
    }

    public Set<String> smembers(final String key) {
        return new Executor<Set<String>>(jedisPool) {
            @Override
            Set<String> execute() {
                return jedis.smembers(key);
            }
        }.setNameKey("smembers", key).getResult();
    }

    public Boolean sismember(final String key, final String member) {
        return new Executor<Boolean>(jedisPool) {
            @Override
            Boolean execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Boolean> result = pipeline.sismember(key, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("sismember", key).getResult();
    }

    public Long srem(final String key, final String member) {
        return new Executor<Long>(jedisPool) {
            @Override
            Long execute() {
                Pipeline pipeline = jedis.pipelined();
                Response<Long> result = pipeline.srem(key, member);
                pipeline.sync();
                return result.get();
            }
        }.setNameKey("srem", key).getResult();
    }


    public void setJedisPool(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    /**
     * 根据key取对象
     */
    public <T> T getObject(final String cacheKey, final Class returnClass) throws IOException {
        String cacheStr = get(cacheKey);
        T object = null;
        if (!Strings.isNullOrEmpty(cacheStr)) {
            object = (T) jsonMapper.readValue(cacheStr, returnClass);
        }
        return object;
    }

    /**
     * 根据key取对象
     */
    public <T> T getObject(final String cacheKey, final TypeReference<T> type) throws IOException {
        String cacheStr = get(cacheKey);
        T object = null;
        if (!Strings.isNullOrEmpty(cacheStr)) {
            object = (T) jsonMapper.readValue(cacheStr, type);
        }
        return object;
    }

    /**
     * redis pipeline
     *
     * @param callback callback
     * @return list result
     */
    public List<Object> batchExecute(final Callback callback) {
        return new Executor<List<Object>>(jedisPool) {
            @Override
            List<Object> execute() {

                Pipeline pipeline = jedis.pipelined();
                callback.callback(pipeline);
                return pipeline.syncAndReturnAll();
            }
        }.setName("batchExecuteWithType").getResult();
    }

    /**
     * redis pipeline
     *
     * @param callback callback
     * @return list result
     */
    public <T> List<T> batchExecuteWithType(final Callback callback) {
        List<T> result = Lists.newArrayList();

        List<Object> list = batchExecute(callback);

        if (list != null && list.size() > 0) {
            for (Object o : list) {
                result.add((T) o);
            }
        }

        return result;
    }


    public <T> List<List<T>> batchExecuteWithType(Callback callback, TypeReference<T> type) {
        List<List<T>> result = Lists.newArrayList();

        List<Object> list = batchExecute(callback);

        if (list != null && list.size() > 0) {
            for (Object o : list) {
                List<T> row = null;
                if (o instanceof LinkedHashSet) {

                    if (o != null && ((LinkedHashSet) o).size() > 0) {
                        row = Lists.newArrayList();
                        for (Object jsonObj : (LinkedHashSet) o) {
                            String json = jsonObj.toString();
                            T cell = null;
                            try {
                                cell = jsonMapper.readValue(json, type);
                            } catch (IOException e) {
                                logger.error(" deserialize error", e);
                            }

                            row.add(cell);
                        }
                    }

                }

                result.add(row);
            }
        }
        return result;
    }


    public interface Callback {
        void callback(Pipeline pipeline);
    }

    abstract class Executor<T> {

        public Jedis jedis;
        public JedisPool jedisPool;
        public String methodName;
        public String __key;

        public Executor(JedisPool jedisPool) {
            this.jedisPool = jedisPool;
            jedis = this.jedisPool.getResource();
        }

        public Executor<T> setName(String methodName) {
            this.methodName = methodName;
            return this;
        }

        public Executor<T> setNameKey(String methodName, String key) {
            this.methodName = methodName;
            this.__key = key;
            return this;
        }

        abstract T execute();

        public T getResult() {
//            if (redisUtilsInterceptor == null) {
//                initialize();
//            }

            T result = null;
            try {
//                redisUtilsInterceptor.startProcess(this.methodName, this.__key);
                result = execute();
            } catch (JedisException e) {
                if (jedis != null) {
                    logger.error(e.getMessage(), e);
                    jedisPool.returnBrokenResource(jedis);
                    jedis = null;
                }
//                redisUtilsInterceptor.exceptionProcess(e.toString());
                throw e;

            } catch (Exception e) {
                logger.error(e.getMessage(), e);
//                redisUtilsInterceptor.exceptionProcess(e.toString());
                throw e;

            } finally {
                if (jedis != null) {
                    jedisPool.returnResource(jedis);
                }
//                redisUtilsInterceptor.endProcess();
            }
            return result;
        }
    }


//    private static void initialize() {
//        try {
//            InputStream is = RedisUtils.class.getClassLoader().getResourceAsStream(interceptorConfig);
//            Properties props = new Properties();
//            props.load(is);
//
//            String redisUtilsInterceptorString = props.getProperty("RedisUtilsInterceptor");
//
//            if (redisUtilsInterceptorString != null && !"".equals(redisUtilsInterceptorString)) {
//                try {
//                    Class RedisUtilsInterceptorClass = Class.forName(redisUtilsInterceptorString);
//                    RedisUtilsInterceptor instance = (RedisUtilsInterceptor) RedisUtilsInterceptorClass.newInstance();
//                    if (instance != null)
//                        redisUtilsInterceptor = instance;
//                } catch (Exception e) {
//                    logger.error("Use the default Interceptor, because error:" + e);
//                    redisUtilsInterceptor = new RedisUtilsDefaultInterceptor();
//                }
//            }
//
//        } catch (Exception e) {
//            logger.error("Use the default Interceptor, because error:" + e);
//            redisUtilsInterceptor = new RedisUtilsDefaultInterceptor();
//        }
//
//    }

//    private static RedisUtilsInterceptor redisUtilsInterceptor;
//    private static String interceptorConfig = "redis-utils-interceptor.properties";
}
